package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import com.practice.gmall.realtime.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_06_TradeOrderRefund
 * Package: com.practice.gmall.realtime.app.dwd
 * Description:
 * <p>
 * 交易域退单事实表
 * 粒度：订单粒度
 * 维度：用户、商品、退单时间、退款原因、来源类型、来源
 * 度量：退单金额、退单数量
 *
 * @Author lzy.ethan@qq.com
 * @Create 2023-02-16 15:03
 */
public class DWD_06_TradeOrderRefund extends BaseSQLApp {
    public static void main(String[] args) {
        new DWD_06_TradeOrderRefund().init(
                3339,
               "DWD_06_TradeOrderRefund",
               2
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {

        MyKafkaUtil.readTopicOdsDb(tEnv, "Dwd_06_DwdTradeOrderRefund");

        tEnv.executeSql(MySQLUtil.getBaseDicLookUp());

        Table orderRefundInfo = tEnv.sqlQuery("" +
                "SELECT   " +
                "    data['id'] id,    " +
                "    data['user_id'] user_id,    " +
                "    data['order_id'] order_id,    " +
                "    data['sku_id'] sku_id,    " +
                "    data['refund_type'] refund_type,    " +
                "    data['refund_num'] refund_num,    " +
                "    data['refund_amount'] refund_amount,    " +
                "    data['refund_reason_type'] refund_reason_type,    " +
                "    data['refund_reason_txt'] refund_reason_txt,    " +
                "    data['create_time'] create_time,    " +
                "    pt,    " +
                "    ts    " +
                "FROM ods_db   " +
                "WHERE `database`='gmall2022'   " +
                "    AND `table`='order_refund_info'   " +
                "    AND `type`='insert' ");

        tEnv.createTemporaryView("order_refund_info", orderRefundInfo);

        Table orderInfo = tEnv.sqlQuery("" +
                "SELECT  " +
                "    `data`['id'] id,   " +
                "    `data`['province_id'] province_id,   " +
                "    `old`   " +
                "FROM ods_db  " +
                "WHERE `database`='gmall2022'  " +
                "    AND `table`='order_info'  " +
                "    AND `type`='update'  " +
                "    AND `old`['order_status']='1002' " +
                "    AND `data`['order_status']='1005'");

        tEnv.createTemporaryView("order_info",orderInfo);

        // join table
        Table orderRefund = tEnv.sqlQuery("" +
                "SELECT" +
                "    ori.order_id, " +
                "    ori.user_id, " +
                "    oi.province_id, " +
                "    ori.sku_id, " +
                "    ori.refund_num, " +
                "    ori.refund_amount, " +
                "    ori.refund_type, " +
                "    dic1.dic_name refund_type_name," +
                "    ori.refund_reason_type," +
                "    dic2.dic_name  refund_reason_type_name," +
                "    ori.refund_reason_txt, " +
                "    ori.create_time order_refund_time, " +
                "    ts   " +
                "FROM" +
                "    order_refund_info ori JOIN order_info oi on ori.order_id = oi.id" +
                "    JOIN base_dic FOR SYSTEM_TIME  AS OF ori.pt as dic1 " +
                "        ON ori.refund_type = dic1.dic_code" +
                "    JOIN base_dic FOR SYSTEM_TIME  AS OF ori.pt as dic2" +
                "        ON  ori.refund_reason_type = dic2.dic_code");

        // create kafka-table to store data
        tEnv.executeSql("" +
                "CREATE TABLE dwd_order_refund( " +
                "    order_id STRING, " +
                "    user_id STRING, " +
                "    province_id STRING, " +
                "    sku_id STRING, " +
                "    refund_num STRING, " +
                "    refund_amount STRING, " +
                "    refund_type STRING, " +
                "    refund_type_name STRING, " +
                "    refund_reason_type STRING, " +
                "    refund_reason_type_name STRING, " +
                "    refund_reason_txt STRING, " +
                "    order_refund_time STRING, " +
                "    ts BIGINT)" +
                MyKafkaUtil.getKafkaSinkDDL("dwd_order_refund"));

        tEnv.executeSql("insert into dwd_order_refund select * from " + orderRefund);
    }
}
